package com.wh.springkafka.wechat.service;

import com.alibaba.fastjson.JSONObject;
import com.wh.springkafka.wechat.config.WechatTemplateProperties;
import com.wh.springkafka.wechat.util.FileUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Optional;

@Slf4j
@Service
public class WechatTemplateServiceImpl {

    private final static String topicName = "wanghao-topic";
    @Autowired
    private WechatTemplateProperties properties;
    @Autowired
    private Producer producer;

    // 获取微信调查模板
    public WechatTemplateProperties.WechatTemplate  getWechatTemplate(){
        List<WechatTemplateProperties.WechatTemplate> templateList = properties.getTemplates();
        Optional<WechatTemplateProperties.WechatTemplate> wechatTemplate
                = templateList.stream().filter(WechatTemplateProperties.WechatTemplate::isActive).findFirst();
        return wechatTemplate.orElse(null);
    }
    // 上报调查问卷填写
    public void  templateReported(JSONObject jsonObject){
        // 推送数据到kafka
        log.info(" templateReported:{}",jsonObject);
        /**
         * 模板格式
         * {
         *     templateId:
         *     result:[]
         * }
         */
        String templateId = jsonObject.getString("templateId");
        /**
         * 1、kafka的Producer是线程安全的，建议多线程复用，如果每个线程都创建则会出现上下文切换的情况，影响kafka
         * 2、kafka Producer的key是个重要的内容：
         *  2.1 我们可以根据key完成Partition的负载均衡。
         *  2.2 合理的key设计，可以让Flink、Spark Streaming之类的分析工具做更快速处理
         * 3、ack-all，kafka层面就已经有了只有一次的消息保障，如果想真的不丢数据，则自己处理异常。
         */
        ProducerRecord<String,Object> record =
                new ProducerRecord<>(topicName,templateId,jsonObject.getJSONArray("result"));



        producer.send(record);

    }
    // 获取调查问卷统计结果
    public JSONObject  templateStatistics(String templateId){
        // 判断略
        if(properties.getTemplateResultType()==0){
            //文件获取
            return FileUtils.readFile2JsonObject(properties.getTemplateResultFilePath()).orElse(null);
        }else{
            //略
        }
        return null;
    }
}
